package com.haiyou.data.common.accessor.redis.redisson.mapping;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.redisson.Redisson;
import org.redisson.RedissonMapCache;
import org.redisson.api.MapCacheOptions;
import org.redisson.api.MapOptions;
import org.redisson.api.RMapCache;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.redisson.connection.CRC16;
import org.springframework.util.ObjectUtils;

import com.google.common.collect.Iterables;

/**
 * RedissonMapCache包装器
 * 
 * 将RMapCache拆分多节点,每个节点上包含128个槽位,默认节点个数16个,设置节点个数时建议使用2^n,最大节点个数为512
 * 
 * @author hong
 */
public class RedissonMapCacheWrapper<K, V> extends RedissonMapCache<K, V> {

	/** 两节点之间的间隔槽数 */
	private static final int NODE_SLOT_BETWEEN = 128;
	
	/** 默认节点个数 */
	public static final int DEFAULT_NODE_SIZE = 1;
	/**
	 * 键类型
	 */
	protected Class<K> keyClass;

	/**
	 * mapping缓存节点
	 */
	List<RMapCache<K, V>> mappings = new ArrayList<>();

	/**
	 * 根据节点数量初始化mapping缓存列表
	 * 
	 * @param redissonClient
	 * @param keyClass
	 * @param keyPrefix
	 * @param SUFFIX
	 * @param mapOptions
	 * @param nodeSize
	 */
	public RedissonMapCacheWrapper(RedissonClient redissonClient, Class<K> keyClass, String key,
			MapCacheOptions<K, V> mapOptions, int nodeSize) {
		super(null, ((Redisson) redissonClient).getCommandExecutor(), key, redissonClient, mapOptions, null);
		this.keyClass = keyClass;
		if (nodeSize == 1) {
			mappings.add(redissonClient.getMapCache(key, mapOptions));
		} else {
			for (int index = 0; index < nodeSize; index++) {
				mappings.add(redissonClient.getMapCache(key + "_" + index, mapOptions));
			}
		}
	}
	
	public RedissonMapCacheWrapper(RedissonClient redissonClient, Class<K> keyClass, String key,
			MapCacheOptions<K, V> mapOptions) {
		this(redissonClient, keyClass, key, mapOptions, DEFAULT_NODE_SIZE);
	}
	
	public RedissonMapCacheWrapper(RedissonClient redissonClient, Class<K> keyClass, String key) {
		this(redissonClient, keyClass, key, null, DEFAULT_NODE_SIZE);
	}
	
	/**
	 * 获得key的crc16数值
	 * 
	 * @param key
	 * @return 0-65535
	 */
	private static int crc16(Object key) {
		if (ObjectUtils.isEmpty(key)) {
			return 0;
		}
		return CRC16.crc16(key.toString().getBytes());
	}

	/**
	 * 获得key值所在的节点
	 * 
	 * @param key
	 * @return
	 */
	private int getNode(Object key) {
		int code = crc16(key);
		int size = mappings.size();
		return slotAt(code, size);
	}

	/**
	 * 获得槽对应的节点
	 * 
	 * @param code
	 * @param size 节点个数
	 * @return
	 */
	private static int slotAt(int code, int nodeSize) {
		// 总槽位
		int total = nodeSize * NODE_SLOT_BETWEEN;
		// code对应的槽位
		int rest = code % total;
		// 槽位对应节点
		return rest / NODE_SLOT_BETWEEN;
	}


	/**
	 * 根据key键返回节点
	 * 
	 * @param key
	 * @return
	 */
	public RMapCache<K, V> getMapping(Object key) {
		if (mappings.size() == 1) {
			return mappings.get(0);
		}
		return mappings.get(getNode(key));
	}

	@Override
	public boolean containsKey(Object key) {
		return getMapping(key).containsKey(key);
	}

	@Override
	public V get(Object key) {
		RMapCache<K, V> mapping = getMapping(key);
		V v = mapping.get(key);
		return v;
	}

	@Override
	public V put(K key, V value, long ttl, TimeUnit unit) {
		return getMapping(key).put(key, value, ttl, unit);
	}

	@Override
	public V put(K key, V value) {
		return getMapping(key).put(key, value);
	}

	@Override
	public V remove(Object key) {
		return getMapping(key).remove(key);
	}

	/**
	 * 各节点批量移除数据
	 */
	@Override
	public long fastRemove(K... keys) {
		if (mappings.size() == 1) {
			return mappings.get(0).fastRemove(keys);
		}

		AtomicLong result = new AtomicLong();
		Map<Integer, Set<K>> groupKeys = groupKeys(keys);

		for(Entry<Integer, Set<K>> entry:groupKeys.entrySet()) {
			Integer node = entry.getKey();
			Set<K> keySet = entry.getValue();
			long count = mappings.get(node).fastRemove(Iterables.toArray(keySet, keyClass));

			result.addAndGet(count);
		}
		return result.get();
	}

	/**
	 * 各节点批量put数据(含过期配置)
	 */
	@Override
	public void putAll(Map<? extends K, ? extends V> map, long ttl, TimeUnit ttlUnit) {
		if (mappings.size() == 1) {
			mappings.get(0).putAll(map, ttl, ttlUnit);
			return;
		}

		Map<Integer, Map<K, V>> groups = groupMap(map);
		for(Entry<Integer, Map<K, V>> entry:groups.entrySet()) {
			getMapping(entry.getKey()).putAll(entry.getValue(), ttl, ttlUnit);
		}
	}

	/**
	 * 各节点批量put数据
	 */
	@Override
	public void putAll(Map<? extends K, ? extends V> map, int batchSize) {
		if (mappings.size() == 1) {
			mappings.get(0).putAll(map);
			return;
		}

		Map<Integer, Map<K, V>> groups = groupMap(map);
		for(Entry<Integer, Map<K, V>> entry:groups.entrySet()) {
			getMapping(entry.getKey()).putAll(entry.getValue());
		}
	}

	/**
	 * 各节点批量get数据
	 */
	@Override
	public Map<K, V> getAll(Set<K> keys) {
		Map<K, V> result = new HashMap<>();
		if (ObjectUtils.isEmpty(keys)) {
			return result;
		}
		if (mappings.size() == 1) {
			return mappings.get(0).getAll(keys);
		}
		// 键分组 key:节点id value:节点下的key列表
		Map<Integer, Set<K>> groups = groupKeys(keys);
		for(Entry<Integer, Set<K>> entry:groups.entrySet()) {
			// 节点下的key列表
			Set<K> ks = entry.getValue();
			// 节点RMapCache
			RMapCache<K, V> cache = mappings.get(entry.getKey());
			Map<K, V> temp = cache.getAll(ks);
			if (ObjectUtils.isEmpty(temp)) {
				continue;
			}
			result.putAll(temp);
		}

		return result;
	}

	/**
	 * 各节点keySet汇总
	 */
	public Set<K> keySet() {
		if (mappings.size() == 1) {
			return mappings.get(0).keySet();
		}

		Set<K> result = new HashSet<>();
		for (RMapCache<K, V> rmap : mappings) {
			result.addAll(rmap.keySet());
		}
		return result;
	}

	/**
	 * 各节点values汇总
	 */
	@Override
	public Collection<V> values() {
		if (mappings.size() == 1) {
			return mappings.get(0).values();
		}
		Set<V> result = new HashSet<>();
		for (RMapCache<K, V> rmap : mappings) {
			result.addAll(rmap.values());
		}
		return result;
	}

	/**
	 * 各节点size汇总
	 */
	public int size() {
		if (mappings.size() == 1) {
			return mappings.get(0).size();
		}
		int result = 0;
		for (RMapCache<K, V> rmap : mappings) {
			result += rmap.size();
		}
		return result;
	}

    @Override
	public RReadWriteLock getReadWriteLock(K key) {
		return getMapping(key).getReadWriteLock(key);
	}
    
    @Override
    public boolean fastPut(K key, V value, long ttl, TimeUnit ttlUnit) {
		return 	getMapping(key).fastPut(key,value,ttl,ttlUnit);
    }

    
	/**
	 * hash 键分组
	 * 
	 * @param keys
	 * @return key:节点id value:键列表
	 */
	private Map<Integer, Set<K>> groupKeys(Iterable<? extends K> keys) {
		Map<Integer, Set<K>> result = new HashMap<>();
		for (K k : keys) {
			// 节点id
			int node = getNode(k);
			// 键列表
			Set<K> set = result.getOrDefault(node, new HashSet<>());
			set.add(k);
			result.put(node, set);
		}
		return result;
	}

	/**
	 * hash 键分组
	 * 
	 * @param keys
	 * @return key:节点id value:键列表
	 */
	private Map<Integer, Set<K>> groupKeys(K... keys) {
		return groupKeys(Arrays.asList(keys));
	}

	/**
	 * 对批量操作的map数据按节点分组
	 * 
	 * @param map
	 * @return key:节点id value:节点内需要操作的map
	 */
	private Map<Integer, Map<K, V>> groupMap(Map<? extends K, ? extends V> map) {
		Map<Integer, Map<K, V>> result = new HashMap<>();
		for (Entry<? extends K, ? extends V> entry : map.entrySet()) {
			// 节点id
			int node = getNode(entry.getKey());
			// 节点下的数据
			Map<K, V> temp = result.getOrDefault(node, new HashMap<>());
			temp.put(entry.getKey(), entry.getValue());
			result.put(node, temp);
		}
		return result;
	}
}
